package message;

import com.google.common.collect.Lists;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;
import java.net.URI;
import java.util.LinkedList;

/**
 * @author Guojt
 * @date 2024/11/12 18:20
 */
public class JMSReceiver implements MessageListener {
    private static String USERNAME = ActiveMQConnection.DEFAULT_USER;
    private static String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
    private static String BROKER = ActiveMQConnection.DEFAULT_BROKER_URL;


    LinkedList<String> messageBuffer = Lists.newLinkedList();

    public JMSReceiver() {
        try {
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKER);
            QueueConnection connection = activeMQConnectionFactory.createQueueConnection();
            connection.start();

            QueueSession session = connection.createQueueSession(false, Session.CLIENT_ACKNOWLEDGE);
            Queue queue = session.createQueue("queue1");
            QueueReceiver receiver = session.createReceiver(queue);
            // 设置消息监听器 为自己
            receiver.setMessageListener(this);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void onMessage(Message message) {
        try {
            if (message.propertyExists("SequenceMarker")) {
                String marker = message.getStringProperty("SequenceMarker");
                // 如果正在启动一个消息组， 清除消息缓冲器
                if (marker.equals("START_SEQUENCE")) {
                    // 我们只检查第一条信息
                    // 由于消息是作为一个组来传递和确认的，任何故障都会导致一个序列消息被标记为正在重新传送，我们并不关心其他消息
                    if (message.getJMSRedelivered()) {
                        processCompensatingTransaction();
                    }
                    messageBuffer.clear();
                }

                // 如果我们正在结束消息组，应处理消息并确认所有的消息已经被传送
                if (marker.equals("END_SEQUENCE")) {
                    //处理消息
                    System.out.println("处理消息");
                    for (String s : messageBuffer) {
                        System.out.println(s + "ssss");
                    }

                    // 确认已经接收到所有的消息
                    message.acknowledge();
                }
            }

            // 如果某一条消息是非标记消息，则保存内容
            if (message instanceof TextMessage) {
                TextMessage msg = (TextMessage) message;
                processInterimMessage(msg.getText());
                if(msg.getText().equals("message3")){
                    throw new RuntimeException("message3"); // 好像消息都是重传的标识
                }
                message.acknowledge();
            }

            System.out.println("waiting for message ...");
        } catch (JMSException e) {
            System.exit(0);
        }
    }

    private void processCompensatingTransaction() {
        // 对前面的消息进行反向处理
        messageBuffer.clear();
    }

    private void processInterimMessage(String msg) {
        // 处理临时消息
        messageBuffer.add(msg);
    }

    public static void main(String[] args) {
        new JMSReceiver();
    }
}
